package ink.feta.task.service.impl;

import ink.feta.task.job.registry.JobAPIProxy;
import ink.feta.task.job.registry.JobRepository;
import ink.feta.task.service.ElasticJobService;
import org.apache.curator.framework.CuratorFramework;
import org.apache.shardingsphere.elasticjob.api.ElasticJob;
import org.apache.shardingsphere.elasticjob.api.JobConfiguration;
import org.apache.shardingsphere.elasticjob.infra.concurrent.BlockUtils;
import org.apache.shardingsphere.elasticjob.infra.pojo.JobConfigurationPOJO;
import org.apache.shardingsphere.elasticjob.lite.internal.election.LeaderService;
import org.apache.shardingsphere.elasticjob.lite.internal.instance.InstanceService;
import org.apache.shardingsphere.elasticjob.lite.internal.schedule.JobRegistry;
import org.apache.shardingsphere.elasticjob.lite.internal.schedule.JobScheduleController;
import org.apache.shardingsphere.elasticjob.lite.internal.schedule.JobScheduler;
import org.apache.shardingsphere.elasticjob.lite.internal.storage.JobNodeStorage;
import org.apache.shardingsphere.elasticjob.lite.lifecycle.domain.JobBriefInfo;
import org.apache.shardingsphere.elasticjob.reg.base.CoordinatorRegistryCenter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import org.springframework.util.ClassUtils;

import javax.annotation.Resource;
import java.util.Collection;

@Service
public class ElasticJobServiceImpl implements ElasticJobService {
    private static Logger logger = LoggerFactory.getLogger(ElasticJobServiceImpl.class);

    @Resource
    CoordinatorRegistryCenter coordinatorRegistryCenter;

    @Resource
    JobAPIProxy jobAPIProxy;

    private JobRegistry jobRegistry = JobRegistry.getInstance();
    private JobRepository jobRepository = JobRepository.getInstance();


    @Override
    public JobConfiguration createJob(JobConfiguration configuration) {
        JobNodeStorage jobNodeStorage = new JobNodeStorage(coordinatorRegistryCenter, configuration.getJobName());
        String className = jobNodeStorage.getJobRootNodeData();
        try {
            //ReflectionUtils.
            ElasticJob elasticJob = (ElasticJob) ClassUtils.forName(className, null).newInstance();
            JobScheduler jobScheduler = new JobScheduler(coordinatorRegistryCenter, elasticJob, configuration);
            JobScheduleController jobScheduleController = jobScheduler.getJobScheduleController();
            jobScheduleController.scheduleJob(configuration.getCron(), null);
            jobRepository.save(configuration.getJobName(), jobScheduler);
        } catch (ReflectiveOperationException e) {
            logger.error("反射类异常：[{}]", e.getMessage(), e);
        }
        return configuration;
    }

    @Override
    public String deleteJob(String jobName) {
        CuratorFramework framework = (CuratorFramework) coordinatorRegistryCenter.getRawClient();
        try {
            String path = "/" + jobName;
            if (framework.checkExists().forPath(path + "/delete") != null) {
                framework.delete().deletingChildrenIfNeeded().forPath(path + "/delete");
                framework.create().forPath(path + "/delete");
            } else {
                framework.create().forPath(path + "/delete");
            }

            InstanceService instanceService = new InstanceService(coordinatorRegistryCenter, jobName);
            LeaderService leaderService = new LeaderService(coordinatorRegistryCenter, jobName);
            while (instanceService.getAvailableJobInstances().size() > 0 || leaderService.hasLeader()) {
                BlockUtils.waitingShortTime();
            }

            if (framework.checkExists().forPath(path) != null) {
                framework.delete().deletingChildrenIfNeeded().forPath(path);
            }
            if (!jobRegistry.isJobRunning(jobName)) {
                return "delete " + jobName + " success";
            }
        } catch (Exception e) {
            e.printStackTrace();
            return "destroy " + jobName + " exception";
        }
        return "destroy " + jobName + " fail";
    }

    @Override
    public JobConfiguration updateJob(String jobName, JobConfiguration configuration) {
        jobAPIProxy.getJobConfigurationAPI().updateJobConfiguration(JobConfigurationPOJO.fromJobConfiguration(configuration));
        return configuration;
    }

    @Override
    public Collection<JobBriefInfo> listJobs() {
        return jobAPIProxy.getJobStatisticsAPI().getAllJobsBriefInfo();
    }

    @Override
    public String pauseJob(String jobName) {
        String status = "pause success";
        jobAPIProxy.getJobOperateAPI().disable(jobName, null);
        return status;
    }

    @Override
    public String resumeJob(String jobName) {
        String status = "resume success";
        jobAPIProxy.getJobOperateAPI().enable(jobName, null);
        return status;
    }
}
